package cn.gwm.flink.streaming.strategy.termal;

import cn.gwm.flink.streaming.beans.termalRunaway.*;
import cn.gwm.flink.streaming.constant.DefaultConstant;
import cn.gwm.flink.streaming.constant.TermalComparisonTypeEnum;
import cn.gwm.flink.streaming.constant.TermalProcessResult;
import cn.gwm.flink.streaming.constant.TermalRunawayTypeEnum;
import cn.gwm.utils.StreamUtil;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.lang.Dict;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.NumberUtil;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.State;

import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**热失控电压异常检测
 * @author GW00283474
 */
public class TermalVoltagePhevStrategy implements TermalStrategy{
    @Override
    public Dict check(State state, long timestamp) {
        try {
            WaringTermalWithStragegyAndFlagPhev result=new WaringTermalWithStragegyAndFlagPhev();
            ListState<WaringTermalDwmWithStragegyPhev> listState=(ListState<WaringTermalDwmWithStragegyPhev>)state;
            Iterator<WaringTermalDwmWithStragegyPhev> recordIterator = listState.get().iterator();
            //如果没有状态值，则返回null
            if(!recordIterator.hasNext()){return null;}
            //信号集合按照tid做正序排列
            List<WaringTermalDwmWithStragegyPhev> recordList = StreamUtil.convertIterator(recordIterator)
                    .sorted(new Comparator<WaringTermalDwmWithStragegyPhev>() {
                        @Override
                        public int compare(WaringTermalDwmWithStragegyPhev o1, WaringTermalDwmWithStragegyPhev o2) {
                            return (int) (o1.getItem_time()- o2.getItem_time());
                        }
                    }).collect(Collectors.toList());
             WaringTermalDwmWithStragegyPhev first=recordList.stream().findFirst().get();
            BeanUtil.copyProperties(first,result);
            BmsTrdStrategys voltageStrategy = first.getBmsTrdStrategies().stream().filter(record -> TermalRunawayTypeEnum.VOLTAGE.getKey().equals(record.getItemType())).findFirst().get();
            //温度状态列表中符合本次处理的所有信号记录集合
            List<WaringTermalDwmWithStragegyPhev> handleRecords = recordList.stream().filter(s -> BooleanUtil.and(s.getItem_time() >= first.getItem_time(), s.getItem_time() < timestamp)).collect(Collectors.toList());
            //单体电压校验
            Dict checkDict=lowVoltRulePattern(handleRecords,voltageStrategy.getComparisonType(), BigDecimal.valueOf(voltageStrategy.getThreshold()),120);
            result.setVoltageWarn(checkDict.get(TermalProcessResult.MATCH,false));
            checkDict.set(TermalProcessResult.CHECK_RECORDS,handleRecords);
            checkDict.set(TermalProcessResult.CHECK_RECORD,result);
            //System.out.println("热失控单电压检测对象：vin="+result.getVin()+"----tid="+result.getTid());
            //更新温度信号状态列表
            Stream<WaringTermalDwmWithStragegyPhev> stream = StreamUtil.convertIterator(listState.get().iterator());
            List<WaringTermalDwmWithStragegyPhev> list=stream.filter(s->NumberUtil.isGreater(new BigDecimal(String.valueOf(s.getItem_time())),new BigDecimal(String.valueOf(first.getItem_time())))).collect(Collectors.toList());
            List<Long> tidList = list.stream().map(WaringTermalDwmWithStragegyPhev::getItem_time).collect(Collectors.toList());
            //System.out.println("剩余有效单电压信号状态列表tids："+tidList);
            ((ListState<WaringTermalDwmWithStragegyPhev>) state).update(list);
            Optional<WaringTermalDwmWithStragegyPhev> min = StreamUtil.convertIterator(((ListState<WaringTermalDwmWithStragegyPhev>) state).get().iterator()).min(new Comparator<WaringTermalDwmWithStragegyPhev>() {
                @Override
                public int compare(WaringTermalDwmWithStragegyPhev o1, WaringTermalDwmWithStragegyPhev o2) {
                    return (int) (o1.getItem_time() - o2.getItem_time());
                }
            });
            if(min.isPresent()){
                //System.out.println("单电压信号状态列表最新最小的tid="+min.get().getTid());
            }else{
                //System.out.println("单电压信号状态列表暂无数据");
            }
            return checkDict;
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 单体电压校验
     * @param records 信号集合
     * @param condition 阈值比对条件
     * @param threshold 阈值
     * @param voltNum 单电压数量
     * @return 校验结果 true  false
     * @throws Exception
     */
    private Dict lowVoltRulePattern(List<WaringTermalDwmWithStragegyPhev> records, int condition, BigDecimal threshold, int voltNum) throws Exception{
        List<Boolean> flags=new ArrayList<>(0);
        //最大单体电压集合
        List<BigDecimal> maxVolts=new ArrayList<>(0);
        //最小单体电压集合
        List<BigDecimal> minVolts=new ArrayList<>(0);
        //最近tid值
        long recentlyTid=Long.MIN_VALUE;
        for (WaringTermalDwmWithStragegyPhev waringTermalDwmWithStragegy: records){
            recentlyTid=NumberUtil.max(waringTermalDwmWithStragegy.getItem_time(),recentlyTid);
            //最大单体电压
            BigDecimal maxVolt=waringTermalDwmWithStragegy.getMaxSingleVolt();
            //最小单体电压
            BigDecimal minVolt=waringTermalDwmWithStragegy.getMinSingleVolt();
            //如果最小电压为默认缺省单体电压值 则不参与计算
            if(DefaultConstant.BMS_SingleCellVolt.compareTo(minVolt)==0){
                continue;
            }else{
                maxVolts.add(maxVolt);
                minVolts.add(minVolt);
                if(TermalComparisonTypeEnum.EQ.getKey()==condition){
                    flags.add(Objects.nonNull(minVolt) && NumberUtil.equals(Convert.toInt(minVolt), threshold.intValue()));
                }
                if(TermalComparisonTypeEnum.LE.getKey()==condition){
                    flags.add(Objects.nonNull(minVolt) && NumberUtil.isLessOrEqual(minVolt, threshold));
                }
                if(TermalComparisonTypeEnum.LESS.getKey()==condition){
                    flags.add(Objects.nonNull(minVolt) && NumberUtil.isLess(minVolt, threshold));
                }
                if(TermalComparisonTypeEnum.GE.getKey()==condition){
                    flags.add(Objects.nonNull(minVolt) && NumberUtil.isGreaterOrEqual(minVolt, threshold));
                }
                if(TermalComparisonTypeEnum.GREATER.getKey()==condition){
                    flags.add(Objects.nonNull(minVolt) && NumberUtil.isGreater(minVolt, threshold));
                }
                if(TermalComparisonTypeEnum.EXCLUDE.getKey()==condition){
                    flags.add(!(Objects.nonNull(minVolt) && NumberUtil.equals(minVolt, threshold)));
                }
            }

        }
        boolean match = false;
        if(CollectionUtil.isNotEmpty(flags)){
            //最终检测结果
            match = flags.stream().allMatch(BooleanUtil::isTrue);
        }
        if(match){
            //最大单体电压
            BigDecimal maxSingleVolt=maxVolts.stream().filter(Objects::nonNull).max(Comparator.comparingDouble(BigDecimal::doubleValue)).get();
            //最小单体电压
            BigDecimal minSingleVolt=minVolts.stream().filter(Objects::nonNull).min(Comparator.comparingDouble(BigDecimal::doubleValue)).get();
            return Dict.of(TermalProcessResult.MATCH, true,
                    TermalProcessResult.MAX_VOLTAGE,maxSingleVolt,
                    TermalProcessResult.MIN_VOLTAGE,minSingleVolt,
                    TermalProcessResult.THRESHOLD_MIN_VOLTAGE,threshold,
                    TermalProcessResult.RECENTLY_TID,recentlyTid,
                    TermalProcessResult.STRATEGY_KEY,TermalRunawayTypeEnum.VOLTAGE.getKey(),
                    TermalProcessResult.STRATEGY_DESC,TermalRunawayTypeEnum.VOLTAGE.getValue());
        }
        return Dict.of(TermalProcessResult.MATCH, false,
                TermalProcessResult.THRESHOLD_MIN_VOLTAGE,threshold,
                TermalProcessResult.RECENTLY_TID,recentlyTid,
                TermalProcessResult.STRATEGY_KEY,TermalRunawayTypeEnum.VOLTAGE.getKey(),
                TermalProcessResult.STRATEGY_DESC,TermalRunawayTypeEnum.VOLTAGE.getValue());
    }
}
